Fox's Git Mirrors
Commit 61af5556362c8bdf7c09a1c01bc84f81629f281b
Parents : 5b767cd
Author : Ivan <e318cbc04468bd574db2b4523dddd710>
Signature : T66BB85Valid, signed by author
Date : 2026-08-10T16:28:36-05:00
feat: improve network interface handling with peer isolation and memory shedding improvements
Changes
16 files changed, 548 insertions(+), 35 deletions(-)
Diff
diff --git a/docs/en/security.md b/docs/en/security.md
index f2bb758f..1fd35c57 100644
--- a/docs/en/security.md
+++ b/docs/en/security.md
@@ -131,7 +131,11 @@ Adaptive baselines use EWMA of once-per-second peak pps/bps. Flood samples are i
Trips emit rate-limited stdout warnings and increment `dos_*` health counters. Full key reference and gate table: [Configuration](configuration.md#dos_protection-go-only). Tests: [Development and testing](development-and-testing.md#dos_protection-tests).
-Handler pool exhaustion always sheds packets (never sync-dispatches on the ingress thread). Priority shedding prefers established link and proof traffic over announce-class floods when slightly over the adaptive trip line.
+Handler pool exhaustion always sheds packets (never sync-dispatches on the ingress thread). Priority shedding prefers established link and proof traffic over announce-class floods when slightly over the adaptive trip line, but that leniency still counts toward the interface's cool-down accounting so sustained abuse of it still escalates instead of running indefinitely.
+
+Memory pressure shedding (heap watermark) enforces immediately in `prevent` and in `auto`, regardless of learning phase. It does not wait for `auto` to arm, since heap exhaustion is a safety valve rather than a flood-learning signal. Explicit `detect` still never blocks, matching its observe-only contract.
+
+Rate, byte, and cool-down accounting run per remote peer as well as per interface. A single sender sharing a listener (a busy TCP/QUIC/VSOCK/I2P accept loop, a UDP socket, or the HTTPS long-poll transport) is capped at half of the interface's effective trip line before its own sub-bucket sheds, so one hostile peer cannot exhaust the whole interface budget and cool down every other peer on it. Peer sub-buckets are bounded and idle ones are pruned so the mitigation cannot itself become a memory-growth vector.
On FreeBSD with sandbox enabled, `SIGHUP` re-execs the daemon so `CapEnter` does not block config reload. Other platforms keep in-process `ReloadInterfaces`.
diff --git a/pkg/interfaces/https.go b/pkg/interfaces/https.go
index a206cbf5..8e24b20e 100644
--- a/pkg/interfaces/https.go
+++ b/pkg/interfaces/https.go
@@ -656,7 +656,7 @@ func (hs *HTTPSServerInterface) handleSend(w http.ResponseWriter, r *http.Reques
return
}
if len(body) > 0 {
- hs.ProcessIncoming(body)
+ hs.ProcessIncomingFrom(body, peerID)
}
w.WriteHeader(http.StatusNoContent)
}
diff --git a/pkg/interfaces/i2p.go b/pkg/interfaces/i2p.go
index 2a334b0c..374afbeb 100644
--- a/pkg/interfaces/i2p.go
+++ b/pkg/interfaces/i2p.go
@@ -101,8 +101,15 @@ type I2PInterfacePeer struct {
wdReset atomic.Bool
done chan struct{}
stopOnce sync.Once
+ peerKey string
}
+// i2pAcceptedPeerSeq gives each accepted I2P peer a unique protect fair-share
+// key. RemoteAddr() on a SAM-tunneled stream commonly resolves to the local
+// SAM bridge socket and is identical across accepted peers, so it cannot be
+// used alone to tell concurrent peers apart.
+var i2pAcceptedPeerSeq atomic.Uint64
+
func NewI2PInterface(name string, cfg *common.InterfaceConfig, ctx *FromConfigContext) (*I2PInterface, error) {
if cfg == nil {
return nil, fmt.Errorf("nil interface config")
@@ -390,6 +397,14 @@ func newI2PInterfacePeerAccepted(parent *I2PInterface, name string, conn net.Con
applyI2PPeerConfig(peer, parent.cfg)
peer.Online = true
_ = setI2PConnTimeouts(conn)
+ seq := i2pAcceptedPeerSeq.Add(1)
+ remote := ""
+ if conn != nil {
+ if ra := conn.RemoteAddr(); ra != nil {
+ remote = ra.String()
+ }
+ }
+ peer.peerKey = fmt.Sprintf("%s#%d", remote, seq)
return peer
}
@@ -712,7 +727,7 @@ func (peer *I2PInterfacePeer) deliverFrame(data []byte) {
peer.parent.RxPackets++
peer.parent.Mutex.Unlock()
}
- peer.ProcessIncoming(data)
+ peer.ProcessIncomingFrom(data, peer.peerKey)
}
func (peer *I2PInterfacePeer) readWatchdog() {
diff --git a/pkg/interfaces/interface.go b/pkg/interfaces/interface.go
index 5d18c8e2..e9e0429a 100644
--- a/pkg/interfaces/interface.go
+++ b/pkg/interfaces/interface.go
@@ -159,13 +159,20 @@ func (i *BaseInterface) GetIFAC() common.IFAC {
}
func (i *BaseInterface) ProcessIncoming(data []byte) {
+ i.ProcessIncomingFrom(data, "")
+}
+
+// ProcessIncomingFrom is ProcessIncoming plus an optional peerKey
+// identifying the remote sender on a shared local interface (for example a
+// listener accepting many client connections). See admitIncomingFrom.
+func (i *BaseInterface) ProcessIncomingFrom(data []byte, peerKey string) {
i.Mutex.Lock()
i.RxBytes += uint64(len(data))
i.RxPackets++
name := i.Name
i.Mutex.Unlock()
- if !admitIncoming(i, name, data) {
+ if !admitIncomingFrom(i, name, data, peerKey) {
return
}
diff --git a/pkg/interfaces/protect_admit.go b/pkg/interfaces/protect_admit.go
index 058eb02e..30560287 100644
--- a/pkg/interfaces/protect_admit.go
+++ b/pkg/interfaces/protect_admit.go
@@ -24,9 +24,21 @@ func ifaceBitrate(iface common.NetworkInterface) int64 {
}
func admitIncoming(iface common.NetworkInterface, name string, data []byte) bool {
+ return admitIncomingFrom(iface, name, data, "")
+}
+
+// admitIncomingFrom is admitIncoming plus an optional peerKey identifying
+// the remote sender on a shared local interface (for example a listener
+// accepting many client connections, or a single UDP socket serving many
+// remote peers). Passing a peerKey gives that sender its own fair-share
+// sub-bucket so it cannot exhaust the whole interface budget and cool down
+// every other peer sharing it. Pass "" for interfaces that are inherently
+// single-peer, where the interface bucket already is the peer bucket.
+func admitIncomingFrom(iface common.NetworkInterface, name string, data []byte, peerKey string) bool {
opts := protect.AdmitOpts{
Bitrate: ifaceBitrate(iface),
Class: protect.PeekPacketClass(data),
+ PeerKey: peerKey,
}
return protect.AdmitPacketOpts(name, len(data), opts).Allow
}
diff --git a/pkg/interfaces/protect_tcp_test.go b/pkg/interfaces/protect_tcp_test.go
index ef2748eb..bc8ba4b2 100644
--- a/pkg/interfaces/protect_tcp_test.go
+++ b/pkg/interfaces/protect_tcp_test.go
@@ -45,6 +45,43 @@ func TestTCPServerProtectConnCap(t *testing.T) {
r4()
}
+func TestSharedListenerPeerIsolationProtectsOtherPeers(t *testing.T) {
+ t.Cleanup(func() { protect.SetDefault(nil) })
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := protect.New(protect.Options{
+ Mode: protect.ModePrevent,
+ MaxPPS: 200,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableAdaptive: true,
+ DisableCoolDown: true,
+ Now: func() time.Time { return clock },
+ })
+ protect.SetDefault(e)
+
+ // Simulates a TCP/QUIC/VSOCK/I2P server-style interface where many
+ // remote peers share one BaseInterface and one protect bucket by name.
+ base := NewBaseInterface("shared-listener", common.IFTypeTCP, true)
+ var delivered atomic.Int64
+ base.SetPacketCallback(func(data []byte, iface common.NetworkInterface) {
+ delivered.Add(1)
+ })
+ pkt := []byte{0x00, 0x00}
+
+ for range 400 {
+ base.ProcessIncomingFrom(pkt, "attacker:1")
+ }
+ before := delivered.Load()
+
+ base.ProcessIncomingFrom(pkt, "friend:1")
+ after := delivered.Load()
+ if after != before+1 {
+ t.Fatalf("quiet peer on the same shared listener must still be delivered despite another peer flooding: before=%d after=%d", before, after)
+ }
+}
+
func TestIfaceChaosProtectFlood(t *testing.T) {
t.Cleanup(func() { protect.SetDefault(nil) })
health.Default.Reset()
diff --git a/pkg/interfaces/quic.go b/pkg/interfaces/quic.go
index 1f8e678b..1f01378f 100644
--- a/pkg/interfaces/quic.go
+++ b/pkg/interfaces/quic.go
@@ -536,11 +536,12 @@ func (qs *QUICServerInterface) SessionCount() int {
}
func (qs *QUICServerInterface) readHDLCLoop(conn net.Conn) {
+ peerKey := conn.RemoteAddr().String()
decoder := newHDLCToggleStreamDecoder(qs.MTU, func(payload []byte) {
if len(payload) == 0 {
return
}
- qs.ProcessIncoming(payload)
+ qs.ProcessIncomingFrom(payload, peerKey)
})
buf := make([]byte, qs.MTU)
for {
diff --git a/pkg/interfaces/tcp.go b/pkg/interfaces/tcp.go
index 4eb53113..c2e5aef3 100644
--- a/pkg/interfaces/tcp.go
+++ b/pkg/interfaces/tcp.go
@@ -728,12 +728,16 @@ func (ts *TCPServerInterface) handleConnection(conn net.Conn) {
}
func (ts *TCPServerInterface) readFramedLoop(conn net.Conn) {
+ peerKey := conn.RemoteAddr().String()
+ onFrame := func(data []byte) {
+ ts.ProcessIncomingFrom(data, peerKey)
+ }
var feed func([]byte)
if ts.kissFraming {
- decoder := newKISSStreamDecoder(ts.MTU, ts.ProcessIncoming)
+ decoder := newKISSStreamDecoder(ts.MTU, onFrame)
feed = decoder.feed
} else {
- decoder := newTCPHDLCStreamDecoder(ts.MTU, ts.ProcessIncoming)
+ decoder := newTCPHDLCStreamDecoder(ts.MTU, onFrame)
feed = decoder.feed
}
buf := make([]byte, ts.MTU)
diff --git a/pkg/interfaces/udp.go b/pkg/interfaces/udp.go
index 624a7aee..92e641a6 100644
--- a/pkg/interfaces/udp.go
+++ b/pkg/interfaces/udp.go
@@ -169,13 +169,22 @@ func (ui *UDPInterface) GetPacketCallback() common.PacketCallback {
}
func (ui *UDPInterface) ProcessIncoming(data []byte) {
+ ui.ProcessIncomingFromAddr(data, "")
+}
+
+// ProcessIncomingFromAddr is ProcessIncoming plus an optional remote address
+// string. A UDP socket is commonly shared by many remote senders, so this
+// gives each sender its own fair-share sub-bucket instead of letting one
+// flooding peer exhaust the whole interface budget and cool down every
+// other peer using the same socket. See admitIncomingFrom.
+func (ui *UDPInterface) ProcessIncomingFromAddr(data []byte, peerKey string) {
ui.Mutex.Lock()
ui.RxBytes += uint64(len(data))
ui.RxPackets++
name := ui.Name
ui.Mutex.Unlock()
- if !admitIncoming(ui, name, data) {
+ if !admitIncomingFrom(ui, name, data, peerKey) {
return
}
@@ -339,7 +348,7 @@ func (ui *UDPInterface) readLoop() {
default:
}
- n, _, err := conn.ReadFromUDP(buffer)
+ n, from, err := conn.ReadFromUDP(buffer)
if err != nil {
ui.Mutex.RLock()
stillOnline := ui.Online
@@ -361,7 +370,11 @@ func (ui *UDPInterface) readLoop() {
return
}
- ui.ProcessIncoming(buffer[:n])
+ peerKey := ""
+ if from != nil {
+ peerKey = from.String()
+ }
+ ui.ProcessIncomingFromAddr(buffer[:n], peerKey)
}
}
diff --git a/pkg/interfaces/vsock.go b/pkg/interfaces/vsock.go
index ec08a71d..63ddde56 100644
--- a/pkg/interfaces/vsock.go
+++ b/pkg/interfaces/vsock.go
@@ -433,11 +433,12 @@ func (vs *VSOCKServerInterface) SessionCount() int {
}
func (vs *VSOCKServerInterface) readHDLCLoop(conn net.Conn) {
+ peerKey := conn.RemoteAddr().String()
decoder := newHDLCToggleStreamDecoder(vs.MTU, func(payload []byte) {
if len(payload) == 0 {
return
}
- vs.ProcessIncoming(payload)
+ vs.ProcessIncomingFrom(payload, peerKey)
})
buf := make([]byte, vs.MTU)
for {
diff --git a/pkg/packet/testdata/handshake_vectors.json b/pkg/packet/testdata/handshake_vectors.json
index bd3c8c76..87c8ba3a 100644
--- a/pkg/packet/testdata/handshake_vectors.json
+++ b/pkg/packet/testdata/handshake_vectors.json
@@ -3,14 +3,14 @@
"vectors": [
{
"name": "announce_h2",
- "raw_hex": "4100abababababababababababababababab52890d7cdb67214026445cd80bc44bbd0086eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646921e506dbf1dd2604ded6feae6a8176dfa28348e1167bff27139ab899d9f461772ea66915f2f5fedf9165fdb7da426006a7a33d9e93b586222cd6eb270b827f93ba9cb029e22be6c24c0be37ad876c11d651f36449e4fbd0800a8a555990d15a154942687ce8ce52ad55958173db57d6db4969090102",
+ "raw_hex": "4100abababababababababababababababab2c2e0380afee31e11536bde5cd54486b00fb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7db93ce1653f497579b2a14501db6619b8c62cfb32a47f2c29c0d6cead57997b0b72ea66915f2f5fedf91616807453bb006a7a4207f3998a0cd1b924ebf90bb7617a214701e6e595db6023be15e9995c50895ad4704463c960de42d01f47b6ccd5eb5ed642622ddc6f01de7483a08fa7b918d83d070102",
"packet_type_name": "ANNOUNCE",
"context_name": "NONE",
"header_type": 1,
"destination_type_name": "SINGLE",
"tree": {
"ok": true,
- "raw_hex": "4100abababababababababababababababab52890d7cdb67214026445cd80bc44bbd0086eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646921e506dbf1dd2604ded6feae6a8176dfa28348e1167bff27139ab899d9f461772ea66915f2f5fedf9165fdb7da426006a7a33d9e93b586222cd6eb270b827f93ba9cb029e22be6c24c0be37ad876c11d651f36449e4fbd0800a8a555990d15a154942687ce8ce52ad55958173db57d6db4969090102",
+ "raw_hex": "4100abababababababababababababababab2c2e0380afee31e11536bde5cd54486b00fb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7db93ce1653f497579b2a14501db6619b8c62cfb32a47f2c29c0d6cead57997b0b72ea66915f2f5fedf91616807453bb006a7a4207f3998a0cd1b924ebf90bb7617a214701e6e595db6023be15e9995c50895ad4704463c960de42d01f47b6ccd5eb5ed642622ddc6f01de7483a08fa7b918d83d070102",
"raw_len": 185,
"flags": 65,
"hops": 0,
@@ -23,22 +23,22 @@
"destination_type_name": "SINGLE",
"context": 0,
"context_name": "NONE",
- "destination_hash": "52890d7cdb67214026445cd80bc44bbd",
+ "destination_hash": "2c2e0380afee31e11536bde5cd54486b",
"transport_id": "abababababababababababababababab",
"data_len": 150,
- "data_hex_prefix": "86eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646"
+ "data_hex_prefix": "fb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7d"
}
},
{
"name": "path_response_announce",
- "raw_hex": "5100abababababababababababababababab52890d7cdb67214026445cd80bc44bbd0b86eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646921e506dbf1dd2604ded6feae6a8176dfa28348e1167bff27139ab899d9f461772ea66915f2f5fedf9165fdb7da426006a7a33d9e93b586222cd6eb270b827f93ba9cb029e22be6c24c0be37ad876c11d651f36449e4fbd0800a8a555990d15a154942687ce8ce52ad55958173db57d6db4969090102",
+ "raw_hex": "5100abababababababababababababababab2c2e0380afee31e11536bde5cd54486b0bfb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7db93ce1653f497579b2a14501db6619b8c62cfb32a47f2c29c0d6cead57997b0b72ea66915f2f5fedf91616807453bb006a7a4207f3998a0cd1b924ebf90bb7617a214701e6e595db6023be15e9995c50895ad4704463c960de42d01f47b6ccd5eb5ed642622ddc6f01de7483a08fa7b918d83d070102",
"packet_type_name": "ANNOUNCE",
"context_name": "PATH_RESPONSE",
"header_type": 1,
"destination_type_name": "SINGLE",
"tree": {
"ok": true,
- "raw_hex": "5100abababababababababababababababab52890d7cdb67214026445cd80bc44bbd0b86eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646921e506dbf1dd2604ded6feae6a8176dfa28348e1167bff27139ab899d9f461772ea66915f2f5fedf9165fdb7da426006a7a33d9e93b586222cd6eb270b827f93ba9cb029e22be6c24c0be37ad876c11d651f36449e4fbd0800a8a555990d15a154942687ce8ce52ad55958173db57d6db4969090102",
+ "raw_hex": "5100abababababababababababababababab2c2e0380afee31e11536bde5cd54486b0bfb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7db93ce1653f497579b2a14501db6619b8c62cfb32a47f2c29c0d6cead57997b0b72ea66915f2f5fedf91616807453bb006a7a4207f3998a0cd1b924ebf90bb7617a214701e6e595db6023be15e9995c50895ad4704463c960de42d01f47b6ccd5eb5ed642622ddc6f01de7483a08fa7b918d83d070102",
"raw_len": 185,
"flags": 81,
"hops": 0,
@@ -51,22 +51,22 @@
"destination_type_name": "SINGLE",
"context": 11,
"context_name": "PATH_RESPONSE",
- "destination_hash": "52890d7cdb67214026445cd80bc44bbd",
+ "destination_hash": "2c2e0380afee31e11536bde5cd54486b",
"transport_id": "abababababababababababababababab",
"data_len": 150,
- "data_hex_prefix": "86eaf8e6e9696dd0ecbe97ce11ce99292b528f2694d3b949b58e8b120c3e1646"
+ "data_hex_prefix": "fb9cefc8268a944dfa54913066de74a81af5fb76eb0bdd5e312ae7e84271ef7d"
}
},
{
"name": "link_request",
- "raw_hex": "020052890d7cdb67214026445cd80bc44bbd00cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd",
+ "raw_hex": "02002c2e0380afee31e11536bde5cd54486b00cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd",
"packet_type_name": "LINKREQUEST",
"context_name": "NONE",
"header_type": 0,
"destination_type_name": "SINGLE",
"tree": {
"ok": true,
- "raw_hex": "020052890d7cdb67214026445cd80bc44bbd00cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd",
+ "raw_hex": "02002c2e0380afee31e11536bde5cd54486b00cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd",
"raw_len": 83,
"flags": 2,
"hops": 0,
@@ -79,7 +79,7 @@
"destination_type_name": "SINGLE",
"context": 0,
"context_name": "NONE",
- "destination_hash": "52890d7cdb67214026445cd80bc44bbd",
+ "destination_hash": "2c2e0380afee31e11536bde5cd54486b",
"data_len": 64,
"data_hex_prefix": "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd"
}
diff --git a/pkg/protect/constants.go b/pkg/protect/constants.go
index 14794811..c8fae21a 100644
--- a/pkg/protect/constants.go
+++ b/pkg/protect/constants.go
@@ -96,4 +96,19 @@ const (
// PersistInterval is how often learning state is flushed to disk.
PersistInterval = 30 * time.Second
+
+ // PeerBudgetFraction is the share of the interface's effective trip
+ // line a single peer may consume before its own sub-bucket sheds,
+ // independent of the interface-wide aggregate. Keeps one hostile peer
+ // on a shared listener from exhausting the whole interface budget.
+ PeerBudgetFraction = 0.5
+
+ // MaxTrackedPeersPerIface bounds per-peer rate sub-buckets so the
+ // mitigation itself cannot become an unbounded-memory DoS vector from
+ // spoofed or churning source identities.
+ MaxTrackedPeersPerIface = 4096
+
+ // PeerIdleEvictAfter prunes a peer sub-bucket that has been quiet this
+ // long, making room for new peers once MaxTrackedPeersPerIface is hit.
+ PeerIdleEvictAfter = 5 * time.Minute
)
diff --git a/pkg/protect/engine.go b/pkg/protect/engine.go
index 2d82873f..79d23222 100644
--- a/pkg/protect/engine.go
+++ b/pkg/protect/engine.go
@@ -45,6 +45,7 @@ type Options struct {
MemorySampleFunc func() uint64
DisableAdaptive bool
DisableCoolDown bool
+ DisablePeerIsolation bool
StorePath string
AutoLearnMinDuration time.Duration
AutoLearnMinSamples int
@@ -60,6 +61,17 @@ type ifaceState struct {
adaptPeakBPS float64
tripAt []time.Time
coolUntil time.Time
+ peers map[string]*peerState
+}
+
+// peerState is a per-remote-peer rate sub-bucket scoped to one interface.
+// It exists so a single sender sharing a listener cannot exhaust the whole
+// interface budget and cool down every other peer on it.
+type peerState struct {
+ window rateWindow
+ tripAt []time.Time
+ coolUntil time.Time
+ lastSeen time.Time
}
type warnKey struct {
@@ -91,6 +103,7 @@ type Engine struct {
memSample func() uint64
disableAdaptive bool
disableCoolDown bool
+ disablePeerIsolation bool
storePath string
autoLearnMinDuration time.Duration
autoLearnMinSamples int
@@ -213,6 +226,7 @@ func New(opts Options) *Engine {
memSample: opts.MemorySampleFunc,
disableAdaptive: opts.DisableAdaptive,
disableCoolDown: opts.DisableCoolDown,
+ disablePeerIsolation: opts.DisablePeerIsolation,
storePath: opts.StorePath,
autoLearnMinDuration: opts.AutoLearnMinDuration,
autoLearnMinSamples: opts.AutoLearnMinSamples,
@@ -389,7 +403,7 @@ func (e *Engine) admitPacket(iface string, nbytes int, opts AdmitOpts) Decision
return Decision{Allow: true}
}
if e.shedMemory.Load() {
- return e.decide(iface, ReasonMemory)
+ return e.decideMemory(iface)
}
now := e.now()
e.mu.Lock()
@@ -398,6 +412,21 @@ func (e *Engine) admitPacket(iface string, nbytes int, opts AdmitOpts) Decision
e.mu.Unlock()
return e.decide(iface, ReasonCoolDown)
}
+ e.mu.Unlock()
+
+ // Peer fair-share runs before the interface aggregate so one hostile
+ // sender on a shared listener (a busy TCP accept loop or a UDP socket
+ // serving many remote peers) trips its own sub-bucket instead of
+ // exhausting the whole interface budget and cooling down every other
+ // peer sharing it.
+ if opts.PeerKey != "" && !e.disablePeerIsolation {
+ if d, deny := e.checkPeer(iface, opts.PeerKey, nbytes, now); deny {
+ return d
+ }
+ }
+
+ e.mu.Lock()
+ st = e.ifaceLocked(iface)
pps, bps := st.window.add(now, nbytes)
sampled := false
samplePPS, sampleBPS := 0.0, 0.0
@@ -418,10 +447,25 @@ func (e *Engine) admitPacket(iface string, nbytes int, opts AdmitOpts) Decision
strictPPS := ppsLimit * 2
strictBPS := bpsLimit * 2
if pps <= strictPPS && bps <= strictBPS {
- if sampled {
- e.maybePromoteOrDrift(iface, samplePPS, sampleBPS)
+ // Claimed link/proof class packets ride out bursts up to 2x the
+ // trip line, but the packet class byte is unauthenticated wire
+ // data any sender controls. Still record the trip and count it
+ // toward interface cool-down so sustained abuse of this
+ // leniency escalates like any other flood instead of being
+ // invisible to health counters and cool-down forever.
+ leniencyReason := ReasonPPS
+ if overBPS && !overPPS {
+ leniencyReason = ReasonBPS
}
- return Decision{Allow: true}
+ d := e.tripCoolDownOnly(iface, leniencyReason)
+ if d.Allow {
+ if sampled {
+ e.maybePromoteOrDrift(iface, samplePPS, sampleBPS)
+ }
+ } else {
+ e.resetDriftLocked()
+ }
+ return d
}
}
e.resetDriftLocked()
@@ -473,10 +517,13 @@ func (e *Engine) resetDriftLocked() {
e.mu.Unlock()
}
-func (e *Engine) tripWithCoolDown(iface string, reason Reason) Decision {
- d := e.decide(iface, reason)
+// accumulateCoolDownTrip records now as a trip timestamp for iface and, once
+// CoolDownTripThreshold trips land within CoolDownTripWindow, arms a full
+// CoolDownDuration cool-down for the interface. Returns true when cool-down
+// was just armed by this call.
+func (e *Engine) accumulateCoolDownTrip(iface string) bool {
if e.disableCoolDown {
- return d
+ return false
}
now := e.now()
e.mu.Lock()
@@ -489,17 +536,165 @@ func (e *Engine) tripWithCoolDown(iface string, reason Reason) Decision {
}
}
st.tripAt = append(kept, now)
- if len(st.tripAt) >= CoolDownTripThreshold {
+ armed := len(st.tripAt) >= CoolDownTripThreshold
+ if armed {
st.coolUntil = now.Add(CoolDownDuration)
st.tripAt = st.tripAt[:0]
- e.mu.Unlock()
+ }
+ e.mu.Unlock()
+ return armed
+}
+
+// decideMemory resolves a shed-memory admission. Heap exhaustion is an
+// absolute safety valve rather than a flood-learning signal, so ModeAuto
+// enforces it immediately even while still in the learning phase and before
+// pps/bps prevention has armed. Explicit ModeDetect stays observe-only,
+// matching its documented contract of never blocking.
+func (e *Engine) decideMemory(iface string) Decision {
+ e.recordTrip(iface, ReasonMemory)
+ if e.mode == ModePrevent || e.mode == ModeAuto {
+ return Decision{Allow: false, Trip: true, Reason: ReasonMemory}
+ }
+ return Decision{Allow: true, Trip: true, Reason: ReasonMemory}
+}
+
+// tripCoolDownOnly records a trip for health counters and interface
+// cool-down accounting without applying decide()'s per-packet enforcement
+// deny. It is used by the prefer-keep leniency band so claimed link/proof
+// traffic can still ride out isolated bursts, while sustained abuse of that
+// leniency still escalates to a full interface cool-down like any other
+// flood, instead of being invisible to metrics and cool-down forever.
+func (e *Engine) tripCoolDownOnly(iface string, reason Reason) Decision {
+ e.recordTrip(iface, reason)
+ if e.accumulateCoolDownTrip(iface) {
e.recordTrip(iface, ReasonCoolDown)
if e.enforcementMode() == ModePrevent {
return Decision{Allow: false, Trip: true, Reason: ReasonCoolDown}
}
- return Decision{Allow: true, Trip: true, Reason: ReasonCoolDown}
}
+ return Decision{Allow: true, Trip: true, Reason: reason}
+}
+
+// peerLocked returns the sub-bucket for peerKey on st, creating one if
+// needed. Must be called with e.mu held. Growth is bounded at
+// MaxTrackedPeersPerIface: idle entries are pruned first, then the least
+// recently seen entry is evicted if still at capacity, so a flood of
+// distinct source identities cannot itself become an unbounded-memory DoS.
+func (e *Engine) peerLocked(st *ifaceState, peerKey string, now time.Time) *peerState {
+ if st.peers == nil {
+ st.peers = make(map[string]*peerState)
+ }
+ ps := st.peers[peerKey]
+ if ps != nil {
+ ps.lastSeen = now
+ return ps
+ }
+ if len(st.peers) >= MaxTrackedPeersPerIface {
+ e.evictStalePeerLocked(st, now)
+ }
+ ps = &peerState{lastSeen: now}
+ st.peers[peerKey] = ps
+ return ps
+}
+
+func (e *Engine) evictStalePeerLocked(st *ifaceState, now time.Time) {
+ var oldestKey string
+ var oldestSeen time.Time
+ for k, ps := range st.peers {
+ if now.Sub(ps.lastSeen) >= PeerIdleEvictAfter {
+ delete(st.peers, k)
+ continue
+ }
+ if oldestKey == "" || ps.lastSeen.Before(oldestSeen) {
+ oldestKey = k
+ oldestSeen = ps.lastSeen
+ }
+ }
+ if len(st.peers) >= MaxTrackedPeersPerIface && oldestKey != "" {
+ delete(st.peers, oldestKey)
+ }
+}
+
+// checkPeer enforces a fair-share budget for a single remote peer sharing
+// iface, independent of the interface-wide aggregate check in admitPacket.
+// It is what stops one hostile peer on a shared listener from exhausting
+// the whole interface budget and cooling down every other peer on it.
+// Returns deny=true when the caller should return the decision immediately
+// instead of continuing to the interface-wide check.
+func (e *Engine) checkPeer(iface, peerKey string, nbytes int, now time.Time) (Decision, bool) {
+ e.mu.Lock()
+ st := e.ifaceLocked(iface)
+ if !e.disableCoolDown {
+ if ps := st.peers[peerKey]; ps != nil && now.Before(ps.coolUntil) {
+ e.mu.Unlock()
+ return e.decide(iface, ReasonCoolDown), true
+ }
+ }
+ ps := e.peerLocked(st, peerKey, now)
+ pps, bps := ps.window.add(now, nbytes)
+ ppsLimit, bpsLimit := st.adapt.tripLine(e.maxPPS, e.maxBPS, e.floorPPS, e.floorBPS)
e.mu.Unlock()
+
+ peerPPSLimit := ppsLimit * PeerBudgetFraction
+ peerBPSLimit := bpsLimit * PeerBudgetFraction
+ if pps <= peerPPSLimit && bps <= peerBPSLimit {
+ return Decision{Allow: true}, false
+ }
+ reason := ReasonPPS
+ if bps > peerBPSLimit && pps <= peerPPSLimit {
+ reason = ReasonBPS
+ }
+ return e.tripPeerCoolDown(iface, peerKey, reason), true
+}
+
+// tripPeerCoolDown mirrors tripWithCoolDown but scopes cool-down state to a
+// single peer bucket instead of the whole interface, so sustained abuse by
+// one peer never blocks the other peers sharing the same local interface.
+func (e *Engine) tripPeerCoolDown(iface, peerKey string, reason Reason) Decision {
+ d := e.decide(iface, reason)
+ if e.disableCoolDown {
+ return d
+ }
+ now := e.now()
+ e.mu.Lock()
+ st := e.ifaceLocked(iface)
+ ps := e.peerLocked(st, peerKey, now)
+ cutoff := now.Add(-CoolDownTripWindow)
+ kept := ps.tripAt[:0]
+ for _, t := range ps.tripAt {
+ if t.After(cutoff) {
+ kept = append(kept, t)
+ }
+ }
+ ps.tripAt = append(kept, now)
+ armed := len(ps.tripAt) >= CoolDownTripThreshold
+ if armed {
+ ps.coolUntil = now.Add(CoolDownDuration)
+ ps.tripAt = ps.tripAt[:0]
+ }
+ e.mu.Unlock()
+ if !armed {
+ return d
+ }
+ e.recordTrip(iface, ReasonCoolDown)
+ if e.enforcementMode() == ModePrevent {
+ return Decision{Allow: false, Trip: true, Reason: ReasonCoolDown}
+ }
+ return Decision{Allow: true, Trip: true, Reason: ReasonCoolDown}
+}
+
+func (e *Engine) tripWithCoolDown(iface string, reason Reason) Decision {
+ d := e.decide(iface, reason)
+ if e.disableCoolDown {
+ return d
+ }
+ if e.accumulateCoolDownTrip(iface) {
+ e.recordTrip(iface, ReasonCoolDown)
+ if e.enforcementMode() == ModePrevent {
+ return Decision{Allow: false, Trip: true, Reason: ReasonCoolDown}
+ }
+ return Decision{Allow: true, Trip: true, Reason: ReasonCoolDown}
+ }
return d
}
@@ -518,7 +713,7 @@ func (e *Engine) AdmitConn(iface string) (Decision, func()) {
return Decision{Allow: true}, noop
}
if e.shedMemory.Load() {
- d := e.decide(iface, ReasonMemory)
+ d := e.decideMemory(iface)
if !d.Allow {
return d, noop
}
@@ -567,7 +762,7 @@ func (e *Engine) AdmitResource(estBytes int64) (Decision, func()) {
}
_ = estBytes
if e.shedMemory.Load() {
- d := e.decide("", ReasonMemory)
+ d := e.decideMemory("")
if !d.Allow {
return d, noop
}
@@ -617,7 +812,7 @@ func (e *Engine) admitSlot(iface string, reason Reason, slot *int, max int) (Dec
return Decision{Allow: true}, noop
}
if e.shedMemory.Load() {
- d := e.decide(iface, ReasonMemory)
+ d := e.decideMemory(iface)
if !d.Allow {
return d, noop
}
diff --git a/pkg/protect/engine_test.go b/pkg/protect/engine_test.go
index 2d212a62..b208faef 100644
--- a/pkg/protect/engine_test.go
+++ b/pkg/protect/engine_test.go
@@ -5,6 +5,7 @@ package protect
import (
"bytes"
+ "fmt"
"strings"
"sync"
"testing"
@@ -332,6 +333,206 @@ func TestObserveMemoryShed(t *testing.T) {
}
}
+func TestMemoryShedEnforcedDuringAutoLearning(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ var heap uint64 = 100
+ e := New(Options{
+ Mode: ModeAuto,
+ SoftMemoryLimit: 1000,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableCoolDown: true,
+ MemorySampleFunc: func() uint64 { return heap },
+ })
+ if e.Phase() != AutoLearning {
+ t.Fatal("expected fresh auto engine to start in learning phase")
+ }
+ heap = 900
+ e.ObserveMemory()
+ if !e.Shedding() {
+ t.Fatal("should shed at 85%")
+ }
+ d := e.AdmitPacket("udp0", 1)
+ if d.Allow {
+ t.Fatal("auto mode must enforce memory shed immediately, even while still learning a baseline")
+ }
+ if e.Phase() != AutoLearning {
+ t.Fatal("memory enforcement should not itself promote the engine to armed")
+ }
+}
+
+func TestMemoryShedStaysObserveOnlyInExplicitDetect(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ var heap uint64 = 900
+ e := New(Options{
+ Mode: ModeDetect,
+ SoftMemoryLimit: 1000,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableCoolDown: true,
+ MemorySampleFunc: func() uint64 { return heap },
+ })
+ e.ObserveMemory()
+ if !e.Shedding() {
+ t.Fatal("should shed at 85%")
+ }
+ d := e.AdmitPacket("udp0", 1)
+ if !d.Allow {
+ t.Fatal("explicit detect mode must never block, even under memory shed")
+ }
+}
+
+func TestPreferKeepLeniencyRecordsTrip(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := New(Options{
+ Mode: ModePrevent,
+ MaxPPS: 2,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableAdaptive: true,
+ DisableCoolDown: true,
+ Now: func() time.Time { return clock },
+ })
+ opts := AdmitOpts{Class: ClassPreferKeep}
+ d1 := e.admitPacket("pk0", 1, opts)
+ d2 := e.admitPacket("pk0", 1, opts)
+ d3 := e.admitPacket("pk0", 1, opts) // pps=3 over MaxPPS(2) but within the 2x leniency band.
+ if !d1.Allow || !d2.Allow || !d3.Allow {
+ t.Fatalf("prefer-keep class should ride out bursts under 2x: %#v %#v %#v", d1, d2, d3)
+ }
+ if e.TripCount(ReasonPPS) == 0 {
+ t.Fatal("prefer-keep leniency use must still be visible in trip counters, not silently invisible")
+ }
+ if !strings.Contains(buf.String(), "trip reason=pps") {
+ t.Fatalf("expected a trip warning even while allowed under leniency, got %q", buf.String())
+ }
+}
+
+func TestTripCoolDownOnlyEscalatesLeniencyAbuse(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := New(Options{
+ Mode: ModePrevent,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ Now: func() time.Time { return clock },
+ })
+ for i := range CoolDownTripThreshold - 1 {
+ d := e.tripCoolDownOnly("pk0", ReasonPPS)
+ if !d.Allow {
+ t.Fatalf("leniency trip %d should still be allowed before threshold: %#v", i, d)
+ }
+ clock = clock.Add(10 * time.Millisecond)
+ }
+ if e.InCoolDown("pk0") {
+ t.Fatal("cool-down should not be armed yet")
+ }
+ d := e.tripCoolDownOnly("pk0", ReasonPPS)
+ if d.Allow || d.Reason != ReasonCoolDown {
+ t.Fatalf("threshold-th sustained leniency trip should arm cool-down: %#v", d)
+ }
+ if !e.InCoolDown("pk0") {
+ t.Fatal("expected iface in cool-down after sustained leniency abuse")
+ }
+ if e.TripCount(ReasonPPS) == 0 {
+ t.Fatal("leniency usage should still be visible in trip counters")
+ }
+}
+
+func TestPeerIsolationShedsOnlyTheFloodingPeer(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := New(Options{
+ Mode: ModePrevent,
+ MaxPPS: 100,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableAdaptive: true,
+ DisableCoolDown: true,
+ Now: func() time.Time { return clock },
+ })
+ // Peer budget is PeerBudgetFraction (0.5) of the 100 pps interface line,
+ // so a single peer sending more than 50 pps should trip its own bucket
+ // well before the interface aggregate itself is threatened.
+ floodOpts := AdmitOpts{PeerKey: "attacker:1"}
+ quietOpts := AdmitOpts{PeerKey: "friend:1"}
+
+ floodBlocked := false
+ for range 80 {
+ d := e.admitPacket("shared0", 1, floodOpts)
+ if !d.Allow {
+ floodBlocked = true
+ }
+ }
+ if !floodBlocked {
+ t.Fatal("expected the flooding peer to eventually be shed by its own sub-bucket")
+ }
+
+ // A different, quiet peer on the same shared interface must be
+ // unaffected by the flooding peer's sub-bucket trip.
+ d := e.admitPacket("shared0", 1, quietOpts)
+ if !d.Allow {
+ t.Fatalf("quiet peer should not be collaterally shed: %#v", d)
+ }
+}
+
+func TestPeerIsolationDisabledFallsBackToSharedBudget(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := New(Options{
+ Mode: ModePrevent,
+ MaxPPS: 5,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableAdaptive: true,
+ DisableCoolDown: true,
+ DisablePeerIsolation: true,
+ Now: func() time.Time { return clock },
+ })
+ opts := AdmitOpts{PeerKey: "attacker:1"}
+ blocked := false
+ for range 20 {
+ d := e.admitPacket("shared0", 1, opts)
+ if !d.Allow {
+ blocked = true
+ }
+ }
+ if !blocked {
+ t.Fatal("expected the shared interface aggregate to still trip when peer isolation is disabled")
+ }
+}
+
+func TestPeerSubBucketEvictionIsBounded(t *testing.T) {
+ health.Default.Reset()
+ var buf bytes.Buffer
+ clock := time.Unix(1_700_000_000, 0)
+ e := New(Options{
+ Mode: ModePrevent,
+ MaxPPS: 1_000_000,
+ WarnWriter: &buf,
+ WarnInterval: time.Hour,
+ DisableAdaptive: true,
+ DisableCoolDown: true,
+ Now: func() time.Time { return clock },
+ })
+ for i := range MaxTrackedPeersPerIface + 50 {
+ _ = e.admitPacket("shared0", 1, AdmitOpts{PeerKey: fmt.Sprintf("peer-%d", i)})
+ }
+ e.mu.Lock()
+ n := len(e.ifaces["shared0"].peers)
+ e.mu.Unlock()
+ if n > MaxTrackedPeersPerIface {
+ t.Fatalf("peer bucket map grew unbounded: %d entries", n)
+ }
+}
+
func TestConfigureFromConfig(t *testing.T) {
t.Cleanup(func() { SetDefault(nil) })
e := ConfigureFromConfig("detect", 0, "", nil)
diff --git a/pkg/protect/packetclass.go b/pkg/protect/packetclass.go
index 4735ec30..64f6886a 100644
--- a/pkg/protect/packetclass.go
+++ b/pkg/protect/packetclass.go
@@ -23,6 +23,14 @@ func (c PacketClass) preferKeep() bool {
type AdmitOpts struct {
Bitrate int64
Class PacketClass
+ // PeerKey identifies the remote sender on a shared local interface (for
+ // example a remote address string). When set, the sender gets its own
+ // fair-share sub-bucket independent of the interface-wide aggregate, so
+ // one hostile peer cannot exhaust the whole interface budget and cool
+ // down every other peer sharing it. Leave empty for interfaces that are
+ // inherently single-peer, where the interface bucket already is the
+ // peer bucket.
+ PeerKey string
}
// wire header bit masks aligned with pkg/transport/wire.go.
diff --git a/reticulum-go.rsm b/reticulum-go.rsm
index 39b34d98..be04dbf9 100644
Binary files a/reticulum-go.rsm and b/reticulum-go.rsm differ
Served by rngit 1.5.2 - Generated in 0.15s